package io.jnet.server.businessprocessor;

import com.jfireframework.baseutil.concurrent.MPMCQueue;
import io.jnet.server.api.ServerListener;
import io.jnet.server.util.ProcessorTask;

public class ConcentrationProcessor implements Runnable
{
    private final MPMCQueue<ProcessorTask> tasks;
    private final ServerListener           serverListener;
    
    public ConcentrationProcessor(MPMCQueue<ProcessorTask> tasks, ServerListener serverListener)
    {
        this.tasks = tasks;
        this.serverListener = serverListener;
    }
    
    @Override
    public void run()
    {
        while (true)
        {
            ProcessorTask task = tasks.take();
            if (task == null)
            {
                // 如果获取到null，则意味着线程被中断，此时应该退出
                break;
            }
            try
            {
                task.getServerChannelContext().push(task.getData(), 0);
            }
            catch (Throwable e)
            {
                serverListener.catchException(e, task.getServerChannelContext());
            }
        }
    }
    
}
